package record

import (
	"bufio"
	"encoding/json"
	"fmt"
	"gamesdk/common/consumer"
	"gamesdk/common/log"
	"gamesdk/common/producer"
	"gamesdk/define"
	"io/ioutil"
	"os"
	"strconv"
	"sync"
	"time"
)

//金币变动总表--使用mgo
type GoldChangeInfo struct {
	ID               int64  //用户ID
	GameUserId       int64  //用户ID
	PlatformId       int64  //平台ID
	UserNickname     string //用户昵称
	ChangeTypeId     int64  //变动类型ID
	ChangeTypeName   int64  //变动类型名称
	ChangeBeforeGold int64  //金币变动前
	ChangeGold       int64  //金币变动数量
	ChangeAfterGold  int64  //金币变动后
	Time             int64  //时间
	Remark           string //变动说明
	CreateTime       int64  //创建时间
	UpdateTime       int64  //更新时间
}

const (
	C_GameClose     = 0 //合
	C_GameTransport = 1 //下注(输)
	C_GameWin       = 2 //赢
)

//游戏变动表(game_change_info)
type GameChangeInfo struct {
	ID                int64  `bson:"ID"`                //主键id
	GameUserId        int64  `bson:"GameUserId"`        //用户ID
	UserNickname      string `bson:"UserNickname"`      //用户昵称
	PlatformId        int64  `bson:"PlatformId"`        //平台ID
	GameId            int64  `bson:"GameId"`            //游戏ID
	RoomId            int64  `bson:"RoomId"`            //房间ID
	GameNum           string `bson:"GameNum"`           //游戏局号
	ChangeTime        int64  `bson:"ChangeTime"`        //变动时间
	ChangeBeforeMoney int64  `bson:"ChangeBeforeMoney"` //变动前金额
	ChangeMoney       int64  `bson:"ChangeMoney"`       //变动金额（收税后）
	ChangeAfterMoney  int64  `bson:"ChangeAfterMoney"`  //变动后金额
	Money             int64  `bson:"Money"`             //操作金额（收税钱）
	Tax               int64  `bson:"Tax"`               //税收比率
	TaxMoney          int64  `bson:"TaxMoney"`          //税收的金额
	Remark            string `bson:"Remark"`            //备注	获胜结果（拉霸记录开奖结果，捕鱼记录鱼种，21点记录玩家点数和庄家点数，百家乐记录本次庄闲等）；
	Type              int    `bson:"Type"`              //类型	下注（输）， 中奖（赢），合
	IntakePool        bool   `bson:"IntakePool"`        //是否进血池 true进 false不进
	IsRobot           bool   `bson:"IsRobot"`           //是否为机器人
}

var g_kafka chan string
var g_rclose chan bool
var g_wclose chan bool
var g_waitGroup sync.WaitGroup
var g_readGroup sync.WaitGroup

func GetDirFileMinName(dir string) int64 {
	files, _ := ioutil.ReadDir(dir)
	min := int64(0)
	for _, file := range files {
		if !file.IsDir() {
			intFileName, err := strconv.ParseInt(file.Name(), 10, 64)
			if err == nil {
				if min == 0 {
					min = intFileName
				} else {
					if intFileName < min {
						min = intFileName
					}
				}
			}
		}
	}
	return min
}

func InitSendKafka() {
	for {
		name := GetDirFileMinName("./record-log")
		if name <= 0 {
			break
		}
		gfile, err := os.OpenFile(fmt.Sprintf("./record-log/%d", name), os.O_RDONLY, 0600)
		if err != nil {
			log.Error(err.Error())
			continue
		}
		gR := bufio.NewReader(gfile)
		var data []string
		for {
			a, _, c := gR.ReadLine()
			if c != nil {
				break
			}
			data = append(data, string(a))
		}
		gfile.Close()
		err = producer.SyncSendMsgs(define.C_Kafka_Game_Change_Info, data)
		if err != nil {
			log.Error("发送kafka失败%s", err.Error())
		} else {
			os.Remove(fmt.Sprintf("./record-log/%d", name))
			log.Info("推送kafka成功")
		}
	}

}
func isJSON(s string) bool {
	var js map[string]interface{}
	return json.Unmarshal([]byte(s), &js) == nil

}

func RunProducer() {
	g_kafka = make(chan string, 10000)
	g_rclose = make(chan bool)
	g_wclose = make(chan bool)

	err := os.Mkdir("./record-log", os.ModePerm)
	if err != nil {
		log.Warn(err.Error())
	}
	InitSendKafka()
	//初始化一个发送协程
	g_waitGroup.Add(1)
	go func() {
		defer g_waitGroup.Done()
		ticker := time.NewTicker(1 * time.Second)
		for {
			select {
			case <-ticker.C:
				name := GetDirFileMinName("./record-log")
				if name <= 0 {
					break
				}
				gfile, err := os.OpenFile(fmt.Sprintf("./record-log/%d", name), os.O_RDONLY, 0600)
				if err != nil {
					log.Error(err.Error())
					break
				}
				gR := bufio.NewReader(gfile)
				var data []string
				for {
					a, _, c := gR.ReadLine()
					if c != nil {
						break
					}
					//fmt.Println(string(a))
					if isJSON(string(a)) {
						data = append(data, string(a))
					} else {
						gfile.Close()
						//log.Warn("%s", string(a))
						break
					}

				}
				gfile.Close()
				err = producer.SyncSendMsgs(define.C_Kafka_Game_Change_Info, data)
				if err != nil {
					log.Error("发送kafka失败%s", err.Error())
				} else {
					os.Remove(fmt.Sprintf("./record-log/%d", name))
					log.Info("推送kafka成功")
				}
			case <-g_wclose:
				return
			}
		}
	}()
	//初始化一个缓存协程
	g_waitGroup.Add(1)
	go func() {
		defer g_waitGroup.Done()
		newOpen := true
		var gFile *os.File
		var gW *bufio.Writer
		ticker := time.NewTicker(1 * time.Second)
		count := 0
		for {
			select {
			case buff := <-g_kafka:
				if newOpen {
					gFile, err = os.OpenFile(fmt.Sprintf("./record-log/%d", time.Now().UnixNano()), os.O_WRONLY|os.O_CREATE, 0600)
					if err != nil {
						log.Error(err.Error())
					} else {
						gW = bufio.NewWriter(gFile)
						newOpen = false
					}
				}
				if newOpen == false {
					fmt.Fprintln(gW, buff)
					count++
					if count >= 50000 {
						gW.Flush()
						gW = nil
						gFile.Close()
						gFile = nil
						newOpen = true
					}
				}
				g_readGroup.Done()
			case <-ticker.C:
				if newOpen == false {
					gW.Flush()
					gW = nil
					gFile.Close()
					gFile = nil
					newOpen = true
					count = 0
				}
			case <-g_rclose:
				if newOpen == false {
					gW.Flush()
					gW = nil
					gFile.Close()
					gFile = nil
					newOpen = true
					count = 0
				}
				producer.Close()
				log.Warn("服务器退出")
				os.Exit(-1)
				return
			}
		}
	}()

}

func CloseSendKafka() {
	defer log.Trace()()
	g_readGroup.Wait()
	g_wclose <- true
	g_rclose <- true
	g_waitGroup.Wait()
	close(g_kafka)
	close(g_wclose)
	close(g_rclose)
}

func SendGameChangeInfoLog(gameUserId int64, userNickname string, platformId, gameId, roomId int64, gameNum string, changeBeforeMoney, changeMoney, changeAfterMoney, money, tax, taxMoney int64, isRobot bool) error {
	defer log.Trace()()
	info := GameChangeInfo{
		GameUserId:        gameUserId,
		UserNickname:      userNickname,
		PlatformId:        platformId,
		GameId:            gameId,
		RoomId:            roomId,
		GameNum:           gameNum,
		ChangeTime:        time.Now().Unix(),
		ChangeBeforeMoney: changeBeforeMoney,
		ChangeMoney:       changeMoney,
		ChangeAfterMoney:  changeAfterMoney,
		Money:             money,
		Tax:               tax,
		TaxMoney:          taxMoney,
		IntakePool:        !isRobot,
		IsRobot:           isRobot,
	}
	body, err := json.Marshal(info)
	if err != nil {
		log.Warn(err.Error())
		return err
	}
	log.Debug("发生%s", string(body))
	g_readGroup.Add(1)
	log.Listen(func() {
		g_kafka <- string(body)
	}, func(fn interface{}) {
		fn.(func())()
	}, fmt.Sprintf("发生上下分%d", gameUserId))
	return nil
}

func ConsumerGameChangeInfoLog(offset int64, f func(partition int32, offset int64, gameChangeInfo *GameChangeInfo, err error)) {
	consumer.RecvMsg(define.C_Kafka_Game_Change_Info, offset, func(partition int32, offset int64, key []byte, value []byte) {
		log.Debug(string(value))
		info := &GameChangeInfo{}
		err := json.Unmarshal(value, info)
		if err != nil {
			log.Error(err.Error())
		}
		f(partition, offset, info, err)
	})
}
